package com.deepglint.sink;

import com.deepglint.beans.SensorReading;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import java.util.ArrayList;
import java.util.HashMap;

/**
 * @author mj
 * @version 1.0
 * @date 2021-11-16 22:31
 */
public class SinkTest_ES {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String inputPath = "C:\\Users\\马军\\Desktop\\Idea-workspace\\flink\\src\\main\\resources\\source.txt";
        DataStream<String> inputStream = env.readTextFile(inputPath);

        // 转换成SensorReading类型
        DataStream<SensorReading> streamMap = inputStream.map(value -> {
            String[] split = value.split(",");
            return new SensorReading(split[0], split[1], new Long(split[2]), new Double(split[3]));
        });

        // 构造hosts
        ArrayList<HttpHost> list = new ArrayList<>();
        list.add(new HttpHost("192.168.150.128", 9200));
        // 将数据存入es
        streamMap.addSink(new ElasticsearchSink.Builder<>(list, new MyElasticSearchSinkFunction()).build());

        env.execute();
    }

    // 自定义ESFunction(写入操作)
    static class MyElasticSearchSinkFunction implements ElasticsearchSinkFunction<SensorReading> {
        @Override
        public void process(SensorReading sensorReading, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
            // 定义写入source
            HashMap<String, String> map = new HashMap<>();
            map.put("id", sensorReading.getId());
            map.put("temp", sensorReading.getTemperature().toString());
            map.put("ts", sensorReading.getTimestamp().toString());

            // 创建请求，作为向es发起的命令
            IndexRequest source = Requests.indexRequest()
                    .index("sensor")
                    .type("readingdata")
                    .source(map);

            // 用index发送请求
            requestIndexer.add(source);
        }
    }
}
